package com.wanji.gateway.processor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.wanji.gateway.event.DataEvent;
import com.wanji.protocol.api.DeviceMessage;
import org.springframework.stereotype.Service;

import java.util.concurrent.Executors;

@Service
public class SensorDataProcessor implements Processor {
    private final Disruptor<DataEvent> disruptor;

    public SensorDataProcessor(DataEventHandler dataEventHandler) {
        this.disruptor = new Disruptor<>(DataEvent::new, 1024, Executors.defaultThreadFactory(), ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWith(dataEventHandler);
        disruptor.start();
    }

    @Override
    public void process(DeviceMessage message) {
        RingBuffer<DataEvent> ringBuffer = disruptor.getRingBuffer();
        long sequence = ringBuffer.next();
        try {
            DataEvent event = ringBuffer.get(sequence);
            event.setDeviceId(message.getDeviceId());
            event.setData(parseSensorData(message));
        } finally {
            ringBuffer.publish(sequence);
        }
    }

    private Object parseSensorData(DeviceMessage message) {
        return message.getData();
    }
}